package com.mima.minaserver.service;
import java.io.IOException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.mima.minaserver.repository.CameraRepository;
import com.mima.minaserver.server.ThreadPoolManager;
import com.mima.minaserver.utils.ParameterUtils;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

import net.sf.ehcache.CacheManager;    

public class RabbitMQListener implements Runnable{  
	private Logger logger = LoggerFactory.getLogger(RabbitMQListener.class);
	
	private Channel channel;
	private ParameterUtils parameterUtils;
	private  CacheManager cacheManager;
	private IotServerMessage iotServerMessage;
	private CameraRepository cameraRepository;
	
	public RabbitMQListener(IotServerMessage iotServerMessage,CacheManager cacheManager,CameraRepository cameraRepository,ParameterUtils parameterUtils) {
		this.parameterUtils = parameterUtils;
		this.cacheManager = cacheManager;
		this.iotServerMessage = iotServerMessage;
		this.cameraRepository = cameraRepository;
		initConnection();
	}
	
	public void initConnection() {
		try {
			logger.info("init RabbitMQ Connection.....");
			ConnectionFactory factory = new ConnectionFactory();
			factory.setHost(parameterUtils.getRabbitmq_host());
	        factory.setUsername(parameterUtils.getRabbitmq_username());
	        factory.setPassword(parameterUtils.getRabbitmq_password());
	        factory.setPort(parameterUtils.getRabbitmq_port());
			Connection connection = factory.newConnection();
			channel = connection.createChannel();
	        logger.info("init RabbitMQ finished!");
		} catch (Exception e) {
			e.printStackTrace();
		}  
	}
	@Override
	public void run() {
		try {
			onMessage();
		}catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	public void onMessage() throws IOException {
		boolean autoAck = false;
		channel.basicConsume(parameterUtils.getQueue_starup_shutdown(), autoAck, "",
		     new DefaultConsumer(channel) {
		         @Override
		         public void handleDelivery(String consumerTag,
		                                    Envelope envelope,
		                                    AMQP.BasicProperties properties,
		                                    byte[] body)
		             throws IOException
		         {
		             String routingKey = envelope.getRoutingKey();
		             String contentType = properties.getContentType();
		             long deliveryTag = envelope.getDeliveryTag();
		             String message = new String(body);
		             JSONObject obj=JSON.parseObject(message);
			         String uuid = obj.getString("uuid");
			         String status = obj.getString("status");
			         String channelValue = obj.getString("channel");
		             logger.info("Received: {" + routingKey + ","+contentType+","+deliveryTag+","+message+"}");
		             ThreadPoolManager.getInstance(parameterUtils.getPool_size()).getExecutorService()
		             .execute(new RabbitMQMessageProcess(iotServerMessage,cameraRepository,cacheManager.getCache("IotServerCache"),
		            		 uuid,Integer.parseInt(status),Integer.parseInt(channelValue)));
		        
		             channel.basicAck(deliveryTag, false);
		         }
		});
	}
}  